#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
RedFire MySQL → PostgreSQL 同步服务
==================================

服务功能:
- 系统服务形式运行同步任务
- 健康检查和监控
- 优雅关闭和故障恢复
- Web API管理接口

Author: RedFire Team
Date: 2025-09-16
Version: v1.0.0
"""

import asyncio
import signal
import logging
import json
import os
import sys
from pathlib import Path
from typing import Dict, Any, Optional
from datetime import datetime
import yaml
from dataclasses import asdict
import uvicorn
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse
from prometheus_client import generate_latest, CONTENT_TYPE_LATEST
import psutil

# 添加项目根目录到路径
current_dir = Path(__file__).parent
project_root = current_dir.parent.parent
sys.path.insert(0, str(project_root))

from databases.sync.mysql_postgresql_sync import (
    MySQLPostgreSQLSyncManager,
    SyncConfig,
    sync_manager_context
)

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class SyncServiceManager:
    """同步服务管理器"""
    
    def __init__(self, config_path: str = None):
        self.config_path = config_path or os.path.join(current_dir, "sync_config.yaml")
        self.config = None
        self.sync_manager = None
        self.running = False
        self.sync_task = None
        self.stats = {
            'start_time': None,
            'last_sync_time': None,
            'total_synced_records': 0,
            'sync_cycles': 0,
            'errors': 0
        }
    
    async def load_config(self) -> Dict[str, Any]:
        """加载配置文件"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                config_data = yaml.safe_load(f)
            
            # 环境变量替换
            config_data = self._replace_env_vars(config_data)
            
            # 获取当前环境配置
            env = os.getenv('REDFIRE_ENV', 'development')
            if env in config_data.get('environments', {}):
                env_config = config_data['environments'][env]
                config_data = self._merge_config(config_data, env_config)
            
            self.config = config_data
            logger.info(f"Configuration loaded from {self.config_path} (env: {env})")
            return config_data
            
        except Exception as e:
            logger.error(f"Failed to load config: {e}")
            raise
    
    def _replace_env_vars(self, data):
        """递归替换环境变量"""
        if isinstance(data, dict):
            return {k: self._replace_env_vars(v) for k, v in data.items()}
        elif isinstance(data, list):
            return [self._replace_env_vars(item) for item in data]
        elif isinstance(data, str) and data.startswith('${') and data.endswith('}'):
            # 解析环境变量 ${VAR_NAME:-default_value}
            env_expr = data[2:-1]
            if ':-' in env_expr:
                var_name, default_value = env_expr.split(':-', 1)
                return os.getenv(var_name, default_value)
            else:
                return os.getenv(env_expr, data)
        else:
            return data
    
    def _merge_config(self, base_config: Dict, env_config: Dict) -> Dict:
        """合并环境特定配置"""
        result = base_config.copy()
        
        def deep_merge(target, source):
            for key, value in source.items():
                if key in target and isinstance(target[key], dict) and isinstance(value, dict):
                    deep_merge(target[key], value)
                else:
                    target[key] = value
        
        deep_merge(result, env_config)
        return result
    
    async def initialize(self):
        """初始化同步服务"""
        await self.load_config()
        
        # 创建同步配置
        sync_config = SyncConfig(
            mysql_config=self.config['mysql'],
            postgresql_config=self.config['postgresql'],
            redis_config=self.config['redis'],
            batch_size=self.config['sync']['batch_size'],
            sync_interval=self.config['sync']['sync_interval'],
            max_retries=self.config['sync']['max_retries'],
            table_mappings={
                k: v['target'] for k, v in self.config['table_mappings'].items()
            }
        )
        
        # 创建同步管理器
        self.sync_manager = MySQLPostgreSQLSyncManager(sync_config)
        await self.sync_manager.initialize()
        
        logger.info("Sync service initialized successfully")
    
    async def start_sync(self):
        """启动同步服务"""
        if self.running:
            logger.warning("Sync service is already running")
            return
        
        self.running = True
        self.stats['start_time'] = datetime.now()
        
        logger.info("Starting MySQL → PostgreSQL sync service...")
        
        # 启动同步任务
        self.sync_task = asyncio.create_task(self._sync_loop())
        
        logger.info("Sync service started successfully")
    
    async def _sync_loop(self):
        """同步循环"""
        while self.running:
            try:
                cycle_start = datetime.now()
                
                # 执行同步
                results = await self.sync_manager.sync_all_tables()
                
                # 更新统计信息
                self.stats['last_sync_time'] = datetime.now()
                self.stats['sync_cycles'] += 1
                
                total_records = sum(
                    r.get('records', 0) for r in results.values() 
                    if r.get('status') == 'success'
                )
                self.stats['total_synced_records'] += total_records
                
                # 统计错误
                errors = sum(
                    1 for r in results.values() 
                    if r.get('status') == 'error'
                )
                self.stats['errors'] += errors
                
                cycle_duration = (datetime.now() - cycle_start).total_seconds()
                
                logger.info(
                    f"Sync cycle {self.stats['sync_cycles']} completed: "
                    f"{total_records} records, {errors} errors, "
                    f"{cycle_duration:.2f}s duration"
                )
                
                # 等待下一个周期
                await asyncio.sleep(self.sync_manager.config.sync_interval)
                
            except asyncio.CancelledError:
                logger.info("Sync loop cancelled")
                break
            except Exception as e:
                logger.error(f"Error in sync loop: {e}")
                self.stats['errors'] += 1
                # 错误时等待更长时间
                await asyncio.sleep(self.sync_manager.config.sync_interval * 2)
    
    async def stop_sync(self):
        """停止同步服务"""
        if not self.running:
            return
        
        logger.info("Stopping sync service...")
        self.running = False
        
        if self.sync_task:
            self.sync_task.cancel()
            try:
                await self.sync_task
            except asyncio.CancelledError:
                pass
        
        if self.sync_manager:
            await self.sync_manager.cleanup()
        
        logger.info("Sync service stopped")
    
    async def get_status(self) -> Dict[str, Any]:
        """获取服务状态"""
        status = {
            'running': self.running,
            'stats': self.stats.copy(),
            'config_loaded': self.config is not None,
            'system_info': {
                'cpu_percent': psutil.cpu_percent(),
                'memory_percent': psutil.virtual_memory().percent,
                'disk_percent': psutil.disk_usage('/').percent,
                'process_id': os.getpid()
            }
        }
        
        if self.stats['start_time']:
            uptime = datetime.now() - self.stats['start_time']
            status['uptime_seconds'] = uptime.total_seconds()
        
        return status
    
    async def force_sync(self, table_name: str = None) -> Dict[str, Any]:
        """强制执行同步"""
        if not self.sync_manager:
            raise RuntimeError("Sync manager not initialized")
        
        if table_name:
            # 同步指定表
            target_table = self.config['table_mappings'].get(table_name, {}).get('target')
            if not target_table:
                raise ValueError(f"Table {table_name} not found in configuration")
            
            result = await self.sync_manager.sync_table(table_name, target_table)
            return {table_name: result}
        else:
            # 同步所有表
            return await self.sync_manager.sync_all_tables()


# 创建全局服务实例
service_manager = SyncServiceManager()

# 创建FastAPI应用
app = FastAPI(
    title="RedFire MySQL → PostgreSQL Sync Service",
    description="实时数据同步服务API",
    version="1.0.0"
)


@app.on_event("startup")
async def startup_event():
    """服务启动事件"""
    try:
        await service_manager.initialize()
        logger.info("FastAPI service started")
    except Exception as e:
        logger.error(f"Failed to start service: {e}")
        raise


@app.on_event("shutdown")
async def shutdown_event():
    """服务关闭事件"""
    await service_manager.stop_sync()
    logger.info("FastAPI service shutdown")


@app.get("/health")
async def health_check():
    """健康检查"""
    try:
        status = await service_manager.get_status()
        return JSONResponse(
            status_code=200,
            content={
                "status": "healthy",
                "timestamp": datetime.now().isoformat(),
                "service": status
            }
        )
    except Exception as e:
        return JSONResponse(
            status_code=500,
            content={
                "status": "unhealthy",
                "error": str(e),
                "timestamp": datetime.now().isoformat()
            }
        )


@app.get("/status")
async def get_status():
    """获取详细状态"""
    status = await service_manager.get_status()
    return JSONResponse(content=status)


@app.post("/sync/start")
async def start_sync(background_tasks: BackgroundTasks):
    """启动同步服务"""
    try:
        background_tasks.add_task(service_manager.start_sync)
        return JSONResponse(content={"message": "Sync service starting"})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/sync/stop")
async def stop_sync():
    """停止同步服务"""
    try:
        await service_manager.stop_sync()
        return JSONResponse(content={"message": "Sync service stopped"})
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/sync/force")
async def force_sync(table_name: Optional[str] = None):
    """强制执行同步"""
    try:
        result = await service_manager.force_sync(table_name)
        return JSONResponse(content=result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/metrics")
async def get_metrics():
    """Prometheus监控指标"""
    try:
        metrics_data = generate_latest()
        return JSONResponse(
            content=metrics_data.decode('utf-8'),
            media_type=CONTENT_TYPE_LATEST
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/config")
async def get_config():
    """获取当前配置"""
    if not service_manager.config:
        raise HTTPException(status_code=404, detail="Configuration not loaded")
    
    # 过滤敏感信息
    safe_config = service_manager.config.copy()
    if 'mysql' in safe_config:
        safe_config['mysql']['password'] = "***"
    if 'postgresql' in safe_config:
        safe_config['postgresql']['password'] = "***"
    if 'redis' in safe_config:
        safe_config['redis']['password'] = "***"
    
    return JSONResponse(content=safe_config)


def setup_signal_handlers():
    """设置信号处理器"""
    def signal_handler(signum, frame):
        logger.info(f"Received signal {signum}")
        asyncio.create_task(service_manager.stop_sync())
    
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)


async def run_service():
    """运行同步服务"""
    setup_signal_handlers()
    
    try:
        await service_manager.initialize()
        await service_manager.start_sync()
        
        # 保持服务运行
        while service_manager.running:
            await asyncio.sleep(1)
            
    except KeyboardInterrupt:
        logger.info("Received keyboard interrupt")
    except Exception as e:
        logger.error(f"Service error: {e}")
    finally:
        await service_manager.stop_sync()


def run_api_server():
    """运行API服务器"""
    config = uvicorn.Config(
        app,
        host="0.0.0.0",
        port=8001,
        log_level="info"
    )
    server = uvicorn.Server(config)
    asyncio.run(server.serve())


if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="RedFire MySQL → PostgreSQL Sync Service")
    parser.add_argument(
        "--mode", 
        choices=["service", "api", "once"], 
        default="service",
        help="运行模式: service(服务模式), api(API模式), once(单次同步)"
    )
    parser.add_argument(
        "--config", 
        help="配置文件路径",
        default=None
    )
    parser.add_argument(
        "--table",
        help="指定要同步的表名(仅在once模式下有效)",
        default=None
    )
    
    args = parser.parse_args()
    
    if args.config:
        service_manager.config_path = args.config
    
    if args.mode == "service":
        # 服务模式：持续运行同步
        asyncio.run(run_service())
    elif args.mode == "api":
        # API模式：启动Web API服务
        run_api_server()
    elif args.mode == "once":
        # 单次模式：执行一次同步后退出
        async def run_once():
            await service_manager.initialize()
            result = await service_manager.force_sync(args.table)
            print(json.dumps(result, indent=2, default=str))
            await service_manager.stop_sync()
        
        asyncio.run(run_once())
